package com.ruoyi.iot;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.stream.CharacterStreamReadingMessageSource;
import org.springframework.messaging.MessageHandler;

import com.ruoyi.iotcommon.constant.MqttTopicConstant;

import jakarta.annotation.Resource;
import lombok.Data;

@Configuration
@Data
public class MqttConfigure {

    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.url}")
    private String hostUrl;

    
	@Resource
	MqttMessageHandler mqttMessageHandler;


    
	@Bean
	public MqttPahoClientFactory mqttClientFactory() {
		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
		MqttConnectOptions options = new MqttConnectOptions();
		options.setServerURIs(new String[]{ hostUrl});
		options.setUserName(username);
		options.setPassword(password.toCharArray());
		options.setAutomaticReconnect(true);
		factory.setConnectionOptions(options);
		return factory;
	}

	// 生产者
	@Bean
	public IntegrationFlow mqttOutFlow() {
		return IntegrationFlow.from(CharacterStreamReadingMessageSource.stdin(),
						e -> e.poller(Pollers.fixedDelay(1000)))
				//.transform(p -> p)
				.handle(mqttOutbound())
				.get();
	}

	@Bean
	public MessageHandler mqttOutbound() {
		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("iot-server-publisher", mqttClientFactory());
		messageHandler.setAsync(true);
		messageHandler.setDefaultTopic("/sys/{prudoctKey}/{deviceKey}/thing/event/{channelKey}/post");
		return messageHandler;
	}

	// 消费者
	@Bean
	public IntegrationFlow mqttInFlow() {
		return IntegrationFlow.from(mqttInbound())
				.handle(mqttMessageHandler)
				.get();
	}

	@Bean
	public MessageProducerSupport mqttInbound() {
		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("iot-server-consumer",
				mqttClientFactory(), MqttTopicConstant.DEFAULT_TOPIC);
		adapter.setCompletionTimeout(5000);
		adapter.setConverter(new DefaultPahoMessageConverter());
		adapter.setQos(0);
		return adapter;
	}
    
}
